#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "../common/ThreadPool.hpp"
#include "../common/UUIDHelper.hpp"
#include "ChannelManager.hpp"
#include <iostream>
#include <memory>
#include <functional>
#include <string>
#include <unordered_map>
#include <cstdio>

namespace micromq
{
    class Connection
    {
    public:
        using ptr = std::unique_ptr<Connection>;
    public:
        Connection(const std::string& serverIp, uint16_t serverPort)
            : _latch(1), _client(_loopThread.startLoop(), muduo::net::InetAddress(serverIp, serverPort), "MicroMQClient")
            , _dispatcher(std::bind(&Connection::onUnknownMessage, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))
            , _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)))
            , _channelManager(std::make_shared<ChannelManager>())
        {
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, 
                std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<CommonResponse>(std::bind(&Connection::onCommonResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<ConsumerResponse>(std::bind(&Connection::onConsumerResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        }

        void connect()
        {
            logInfo("client connecting...");
            _client.connect();
            _latch.wait();
        }

        Channel::ptr openChannel()
        {
            Channel::ptr channel = _channelManager->addChannel(_conn, _codec);
            if(channel->openChannel() == false)
            {
                logInfo("open channel faild...");
                return nullptr;
            }
            return channel;
        }

        void closeChannel(const Channel::ptr& channel)
        {
            channel->closeChannel();
            _channelManager->removeChannel(channel->getChannelId());
        }

        ~Connection()
        {}
    private:
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected()) {
                logInfo("client connection established...");
                _conn = conn;
                _latch.countDown();
            } else {
                logInfo("connection closed...");
            }
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, 
            const std::shared_ptr<google::protobuf::Message>& message, muduo::Timestamp)
        {
            logInfo("unknown message: %s", message->GetTypeName().c_str());
            conn->shutdown();
        }

        void onConsumerResponse(const muduo::net::TcpConnectionPtr& conn, const ConsumerResponsePtr& resp, muduo::Timestamp)
        {
            Channel::ptr channel = _channelManager->getChannel(resp->channel_id());
            if(channel.get() == nullptr)
            {
                logInfo("channel can not found...");
                return;
            }
            ThreadPool::getInstance()->submit([channel, resp]() {
                channel->consumeMessage(resp);
            });
        }

        void onCommonResponse(const muduo::net::TcpConnectionPtr& conn, const CommonResponsePtr& resp, muduo::Timestamp)
        {
            Channel::ptr channel = _channelManager->getChannel(resp->channel_id());
            if(channel.get() == nullptr)
            {
                logInfo("channel can not found...");
                return;
            }
            channel->putResponse(resp);
        }
    private:
        muduo::CountDownLatch _latch;
        muduo::net::EventLoopThread _loopThread;
        muduo::net::TcpConnectionPtr _conn;
        muduo::net::TcpClient _client;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;

        ChannelManager::ptr _channelManager;
    };
}